package task

import (
	"crypto/sha1"
	"encoding/hex"
	"encoding/json"
	"fmt"
	kafkamodel "gitee.com/zaper/kafka-service/model"
	"gitee.com/zaper/kafka-service/service/producer"
	"gitee.com/zaper/kafka-service/service/producer/producer_model"
	logger "gitee.com/zaper/private-logger"
	"gitee.com/zaper/service-config/constant"
	"gitee.com/zaper/service-config/service/model"
	"math/rand"
	"runtime/debug"
	"strings"
	"sync"
	"time"
)

type Task struct {
	TaskId      string
	GroupName   string
	TaskVersion string
	ServiceList []string
	Message     []model.ConfigKeyValue
	Stop        chan struct{}
	sw          sync.Mutex
}

func NewTask(groupName string, serviceList []string, message []model.ConfigKeyValue) *Task {
	return &Task{
		TaskId:      parseDataSha(message),
		GroupName:   groupName,
		Message:     message,
		ServiceList: serviceList,
		Stop:        make(chan struct{}),
	}
}

func (task *Task) ParseKey() []byte {
	//return []byte(strings.Join(task.ServiceList, ","))
	return []byte(task.GroupName)
}

func (task *Task) ParseValue() []byte {
	var messModel model.ConfigModel
	messModel.UUID = task.TaskId
	messModel.ServiceList = strings.Join(task.ServiceList, ",")
	messModel.ConfigKeyValueList = task.Message
	v, err := json.Marshal(messModel)
	if err != nil {
		return nil
	}
	return v
}

func RemoveTaskService(serviceName string, task *Task) {
	task.sw.Lock()
	defer task.sw.Unlock()
	var serviceList []string
	for _, srv := range task.ServiceList {
		if serviceName == srv {
			continue
		}
		serviceList = append(serviceList, srv)
	}
	task.ServiceList = serviceList
}

func (task *Task) execTask() {
	defer func() {
		e := recover()
		if e != nil {
			logger.Errorf("exec task err, err is %v, stack is %s", e, debug.Stack())
			return
		}
	}()
	defer func() {
		task.postTaskEndHandler()
	}()
	taskGroup := GetTaskGroup()
	currTask := taskGroup.GetGroupCurrentTask(task.GroupName)
	logger.Infof("New task is: %s, Current task is %v", task.ToString(), currTask)
	if currTask != nil {
		// 新任务和当前正在运行的任务相同，忽略新任务。反之，则结束当前任务
		if currTask.TaskId == task.TaskId {
			logger.Warn("new task and current task same, return")
			return
		} else {
			currTask.Stop <- struct{}{}
		}
	}
	taskGroup.SetGroupCurrentTask(task.GroupName, task)
	notifyServiceConfig(task)
	ticker := time.NewTicker(120 * time.Second)
	for {
		select {
		case <-ticker.C:
			if rt := notifyServiceConfig(task); rt {
				return
			}
		case <-task.Stop:
			logger.Warn("Current go thread exit ")
			return
		}
	}
}

func notifyServiceConfig(task *Task) bool {
	if len(task.ServiceList) == 0 {
		// 客户端收到通知后回主动清理serviceList，只有所有服务都上报成功serviceList就会清空。才会结束任务。
		// 及时消息发送成功，但也有可能客户端处理失败，因此需要再继续发送。
		logger.Infof("all services finish!!")
		return true
	}
	pdu := producer.NewProducerAgent()
	pdu.ProducerBind(&producer_model.ProducerModel{
		Topic: constant.ConfigSubKafkaTop,
	})
	err := pdu.ProducerMess(&kafkamodel.MessModel{
		Key:   task.ParseKey(),
		Value: task.ParseValue(),
	})
	if err != nil {
		logger.Errorf("ProducerMess fail, err is %s", err)
		return false
	}
	logger.Info(" success produce ")
	return false
}

func parseDataSha(message []model.ConfigKeyValue) string {
	hashes := sha1.New()
	var dataArr []string
	for _, kv := range message {
		dataArr = append(dataArr, kv.Value)
	}
	hashes.Write([]byte(strings.Join(dataArr, ",")))
	hash := hashes.Sum(nil)
	return hex.EncodeToString(hash)
}

// 正常结束后需要清理任务信息，否则下次启动任务通道会卡死
// 连续变更配置，前面的任务被清理掉之后，也需要清理任务，否则下次启动任务也会卡死
// 但是循环通知是异步，如果defer晚于外层设置当次任务id时候，会将新任务覆盖
func (task *Task) postTaskEndHandler() {
	manager := GetTaskGroup()
	current := manager.GetGroupCurrentTask(task.GroupName)
	if current.TaskId != task.TaskId {
		// 说明当前任务id已经变更，已经是新任务了，此时当前任务不需要至空
		return
	}
	manager.SetGroupCurrentTask(task.GroupName, nil)
}

func (task *Task) ToString() string {
	return fmt.Sprintf("GroupName is %s TaskId is %s, ServiceList is %s, Message is %s", task.GroupName,
		task.TaskId, task.ServiceList, task.Message)
}

func RandTaskID() int {
	return rand.Intn(999) + 1
}
